package util;

import java.io.IOException;
import java.security.KeyStore.Entry;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;


import org.antlr.runtime.tree.CommonTree;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.idx.api.IndexManipulator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.idx.exp.Comparison.Operator;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.generated.master.table_jsp;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.regionserver.IdxRegion;
import org.apache.hadoop.hbase.util.Bytes;


import conf.UserConfigure;

public class MetaDataHandler {

	private static Configuration configuration = UserConfigure.configuration;
//	private static Configuration configuration = null;
//	static{
//		configuration = HBaseConfiguration.create();
//		configuration.addResource(new Path("./src/conf/user-connect.xml"));
////		configuration.set("hbase.master", "192.9.200.172:60000");
////		configuration.set("hbase.zookeeper.quorum","192.9.200.10");  
////		configuration.set("hbase.zookeeper.property.clientPort", "2181");    
////		configuration.set("hbase.HServerAddress", "192.9.200.172:60000");
//	}
////	static{
////		configuration = HBaseConfiguration.create();
////		configuration.addResource(new Path("src/hbase-site.xml"));
////		configuration.setClass(HConstants.REGION_IMPL, IdxRegion.class, IdxRegion.class);
////		configuration.set("hbase.hregion.memstore.flush.size", "262144");
////	}
	/**
	 * init the HbaseAdmin
	 * @return
	 * @throws MasterNotRunningException
	 * @throws ZooKeeperConnectionException
	 */
	private static HBaseAdmin admin = null;
	static{
		try {
			admin = new HBaseAdmin(configuration);
		} catch (MasterNotRunningException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ZooKeeperConnectionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
/*	
	private HBaseAdmin setAdmin() throws MasterNotRunningException, ZooKeeperConnectionException{
		HBaseAdmin admin = null;
		admin = new HBaseAdmin(configuration);
		return admin;
	}
*/

	/**
	 * ourselves'
	 * check whether the table exists
	 * @param tab_name
	 * @return
	 * @throws IOException 
	 */
	public static boolean isTableExistsOld(String tab_name){
		try {
			return admin.tableExists(tab_name);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return false;
	}
	

	/**
	 * check whether the table is enabled;
	 * when execute 'drop table', before 'delete table',it will check this,so you donnot need to call this method
	 * but test is wrong, so you also need to disable the table before drop it
	 * @param tab_name
	 * @return
	 * @throws IOException 
	 */
	public static boolean isTableEnabled(String tab_name) throws IOException{
		return admin.isTableEnabled(tab_name);
	}
	/**
	 * create a table, given tab_name, list of column families
	 * @param tab_name
	 * @param families
	 * @throws IOException
	 */
	public static void createTable(String tab_name,ArrayList<String> families) throws IOException{
		HTableDescriptor descriptor = new HTableDescriptor(tab_name);
		for (String family : families) {
//			if (!descriptor.hasFamily(family.getBytes())) {
				descriptor.addFamily(new HColumnDescriptor(family));
//			}
//			else {
//				System.out.println("锟斤拷锟斤拷锟截革拷锟斤拷锟斤拷锟斤拷锟斤拷");
//				return;
//			}
		}
		admin.createTable(descriptor);
	}
	/**
	 * drop the table,given the tab_name
	 * @param tab_name
	 * @throws IOException
	 */
	public static void dropTable(String tab_name) throws IOException{
		disableTable(tab_name);
		admin.deleteTable(tab_name);
	}
	/**
	 * enable the table
	 * @param tab_name
	 * @throws IOException
	 */
	public static void enableTable(String tab_name) throws IOException{
		admin.enableTable(tab_name);
	}
	/**
	 * disable the table
	 * @param tab_name
	 * @throws IOException
	 */
	public static void disableTable(String tab_name) throws IOException{
		admin.disableTable(tab_name);
	}
	/**
	 * check whether the table 'tab_name' contains the column family 'cofFamily'
	 * @param tab_name
	 * @param family
	 * @return
	 * @throws IOException 
	 */
	public static boolean hasColumnFamily(String tab_name, String colFamily) throws IOException{
		HTableDescriptor descriptor = null;
		descriptor = admin.getTableDescriptor(tab_name.getBytes());
		return descriptor.hasFamily(colFamily.getBytes());
	}

	/**
	 * get all the column families that contained by the table
	 * @param tab_name
	 * @return
	 * @throws IOException
	 */
	public static ArrayList<String> getColumnFamily(String tab_name){
		ArrayList<String> columnFamilies = new ArrayList<String>();
		HTableDescriptor descriptor = null;
		Set<byte[]> familySet = null;
		try {
			descriptor = admin.getTableDescriptor(tab_name.getBytes());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		familySet = descriptor.getFamiliesKeys();
		Iterator<byte[]> it = familySet.iterator();
		while (it.hasNext()) {
			byte[] bs = (byte[]) it.next();
			columnFamilies.add(new String(bs));
		}
		return columnFamilies;
	}
	/**
	 * add a column family to the table
	 * @param tab_name
	 * @param colFamily
	 * @throws IOException 
	 */
	public static void addColumnFamily(String tab_name, String colFamily) throws IOException{
		HColumnDescriptor cFDescriptor = new HColumnDescriptor(colFamily);
		admin.disableTable(tab_name);
		admin.addColumn(tab_name, cFDescriptor);
		admin.enableTable(tab_name);
	}
	/**
	 * delete a column family
	 * @param tab_name
	 * @param colFamily
	 * @throws IOException 
	 */
	public static void deleteColumnFamily(String tab_name, String colFamily) throws IOException{
		admin.disableTable(tab_name);
		admin.deleteColumn(tab_name, colFamily);
		admin.enableTable(tab_name);
	}
	
	public static void hasColum(String tab_name, String colFamily, String column){
		
	}
	/**
	 * get all the columns contained in the table
	 * @param tab_name
	 * @return
	 * @throws IOException
	 */
	public static ArrayList<String> getColumns(String tab_name) throws IOException{
		ArrayList<String> columns = new ArrayList<String>();
		HTable table = new HTable(configuration,tab_name);
		Scan sc = new Scan();
		ResultScanner rs = table.getScanner(sc);
		String column = "";
		for (Result r : rs) {
			for (KeyValue kv : r.raw()) {
				column = new String(kv.getFamily()) + ":" + new String(kv.getQualifier());
				if (!columns.contains(column)) {
					columns.add(column);
				}
			}
		}
		return columns;
	}
	
	/**
	 * ourselves'
	 * get all the columns of some table's some family
	 * @param tab_name
	 * @param colFamily
	 * @return
	 */
	public static ArrayList<String> getColumnsOld(String tab_name, String colFamily) {
		ArrayList<String> columns = new ArrayList<String>();
		HTable table=null;
		try {
			table = new HTable(configuration,tab_name);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return null;
		}
//		Scan sc = new Scan();
		ResultScanner rs=null;
		try {
			rs = table.getScanner(colFamily.getBytes());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return null;
		}
		String column = "";
		for (Result r : rs) {
			for (KeyValue kv : r.raw()) {
				column = new String(kv.getQualifier());
				if (!columns.contains(column)) {
					columns.add(column);
				}
			}
		}
		return columns;
	}
	
	/**
	 * get all the columns call the api of chengxu
	 * @param tab_name
	 * @param colFamily
	 * @return
	 */

	/**
	 * get the count of the rows
	 * @param tab_name
	 * @return
	 * @throws IOException
	 */
	public static int getRowCount(String tab_name) throws IOException{
		int count = 0;
		HTable table = new HTable(configuration,tab_name);
		Scan sc = new Scan();
		sc.setFilter(new FirstKeyOnlyFilter());
		ResultScanner rs = table.getScanner(sc);
		Iterator<Result> it = rs.iterator();
		while (it.hasNext()) {
			it.next();
			count++;
		}
		return count;
	}
	/**
	 * add one cell's data
	 * @param tab_name
	 * @param rowKey
	 * @param colFamily
	 * @param column
	 * @param value
	 * @throws IOException
	 */
	public static void addData(String tab_name, String rowKey, String colFamily, String column, String value) 
		throws IOException{
		HTable table = new HTable(configuration,tab_name);
		Put put = new Put(rowKey.getBytes());
		put.add(colFamily.getBytes(),column.getBytes(),value.getBytes());
		table.put(put);
		table.clearRegionCache();
		table.close();
	}
	/**
	 * delete one row's data
	 * @param tab_name
	 * @param rowKey
	 * @throws IOException
	 */
	public static void deleteRowData(String tab_name, String rowKey) throws IOException{
		HTable table = new HTable(configuration,tab_name);
		Delete del = new Delete(rowKey.getBytes());
		table.delete(del);
	}
	/**
	 * delete one cell's data
	 * @param tab_name
	 * @param rowKey
	 * @param colFamily
	 * @param column
	 * @throws IOException
	 */
	public static void deleteCellData(String tab_name, String rowKey, String colFamily, String column) 
		throws IOException{
		HTable table = new HTable(configuration,tab_name);
		Delete del = new Delete(rowKey.getBytes());
		del.deleteColumn(colFamily.getBytes(), column.getBytes());
		table.delete(del);
	}
	/**
	 * get one row's data, all the columns in this row
	 * @param tab_name
	 * @param rowKey
	 * @return
	 * @throws IOException
	 */
	
	
	/**
	 * Scan the table to get all the data, but not return the htable, it returns the ResultScanner
	 * @param tableName
	 * @return
	 */
	public static ResultScanner getResultScanner(String tableName){
		HTable table = null;
		try {
			table = new HTable(configuration, tableName);
			Scan sc = new Scan();
			ResultScanner rs = null;
			rs = table.getScanner(sc);
			return rs;
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;
	}
	
	public static void getAllDataTest(String tab_name) throws IOException{
		ArrayList<String> attribute = new ArrayList<String>();
		HTable table = new HTable(configuration,tab_name);
		Scan scan = new Scan();
		ResultScanner rs = table.getScanner(scan);
		
		for(Result r : rs){
			for(KeyValue kv : r.raw()){
				System.out.println("RowKey:"+new String(kv.getRow())+", family:"+new String(kv.getFamily())+",qualifier:"
						+new String(kv.getQualifier())+",value:"
						+new String(kv.getValue()));
			}
			System.out.println();
		}
//		return null;
	}
	public static Hashtable<String, Hashtable<String, String>> getAllDataToHashTable(String tab_name) throws IOException{
//		ArrayList<String> attribute = new ArrayList<String>();
		Hashtable<String, Hashtable<String, String>> ht = new Hashtable<String, Hashtable<String, String>>();
		HTable table = new HTable(configuration,tab_name);
		Scan scan = new Scan();
		ResultScanner rs = table.getScanner(scan);
		
		for(Result r : rs){
			for(KeyValue kv : r.raw()){
				String rowkey = new String(kv.getRow());
				String cf = new String(kv.getFamily());
				String qf = new String(kv.getQualifier());
				String vl = new String(kv.getValue());
				if(ht.contains(rowkey))
				{
					ht.get(rowkey).put(cf+":"+qf, vl);
				}
				else
				{
					Hashtable<String, String> tmp = new Hashtable<String, String>();
					tmp.put(cf+":"+qf, vl);
					ht.put(rowkey, tmp);
				}
//				System.out.println("RowKey:"+new String(kv.getRow())+", family:"+new String(kv.getFamily())+",qualifier:"
//						+new String(kv.getQualifier())+",value:"
//						+new String(kv.getValue()));
			}
//			System.out.println();
		}
//		return null;
		return ht;
	}
	
	public static Hashtable<String, Hashtable<String, String>> getPartDataToHashTable(String tab_name, String startRow, String stopRow) throws IOException{
//		ArrayList<String> attribute = new ArrayList<String>();
		Hashtable<String, Hashtable<String, String>> ht = new Hashtable<String, Hashtable<String, String>>();
		HTable table = new HTable(configuration,tab_name);
		Scan scan = new Scan();
		if(startRow!=null)
		{
			scan.setStartRow(startRow.getBytes());
		}
		if(stopRow!=null)
		{
			scan.setStopRow(stopRow.getBytes());
		}
		ResultScanner rs = table.getScanner(scan);
		
		for(Result r : rs){
			for(KeyValue kv : r.raw()){
				String rowkey = new String(kv.getRow());
				String cf = new String(kv.getFamily());
				String qf = new String(kv.getQualifier());
				String vl = new String(kv.getValue());
				if(ht.contains(rowkey))
				{
					ht.get(rowkey).put(cf+":"+qf, vl);
				}
				else
				{
					Hashtable<String, String> tmp = new Hashtable<String, String>();
					tmp.put(cf+":"+qf, vl);
					ht.put(rowkey, tmp);
				}
//				System.out.println("RowKey:"+new String(kv.getRow())+", family:"+new String(kv.getFamily())+",qualifier:"
//						+new String(kv.getQualifier())+",value:"
//						+new String(kv.getValue()));
			}
//			System.out.println();
		}
//		return null;
		return ht;
	}

	/**
	 * unuseful,get the info of a table's families
	 * @param tab_name
	 * @throws IOException
	 */
	public static void getFamilyInfo(String tab_name) throws IOException{
		HTableDescriptor descriptor = admin.getTableDescriptor(tab_name.getBytes());
		HColumnDescriptor[] hcd = descriptor.getColumnFamilies();
		Map<ImmutableBytesWritable, ImmutableBytesWritable> values = hcd[0].getValues();
		Iterator<java.util.Map.Entry<ImmutableBytesWritable, ImmutableBytesWritable>> it = values.entrySet().iterator();
		String a = "", b = "";
		while (it.hasNext()) {
			Map.Entry<org.apache.hadoop.hbase.io.ImmutableBytesWritable, org.apache.hadoop.hbase.io.ImmutableBytesWritable> entry = (Map.Entry<org.apache.hadoop.hbase.io.ImmutableBytesWritable, org.apache.hadoop.hbase.io.ImmutableBytesWritable>) it
					.next();
			a = new String(entry.getKey().get());
			b = new String(entry.getValue().get());
			System.out.println(a+"\t"+b);
		}
	}
	
	public static void createIndex(String tab_name, String colFamily, String column, boolean isUnique) {
		try {
			IndexManipulator.CreateIndex(admin, tab_name.getBytes(), colFamily.getBytes(), column.getBytes(), isUnique);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public static void showIndex(String tab_name, String colFamily, List<byte[]> coList, List<Boolean> isUnique) {
		IndexManipulator.ShowIndex(tab_name.getBytes(), colFamily.getBytes(), coList, isUnique);
	}
	
	public static void dropIndex(String tab_name, String colFamily, String column) {
		try {
			IndexManipulator.DropIndex(admin, tab_name.getBytes(), colFamily.getBytes(), column.getBytes());
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	



	public static Configuration getConf(){
		return configuration;
	}
	
	//test whether the client could connect the server
	public static boolean testConnect() {
		// TODO Auto-generated method stub		
		return isTableExistsOld("-ROOT-");
	}


}
